
/*****************************************************************************************
 * @file  Table.java
 *
 * @author   John Miller
 * @author   Michael Cotterell
 * @author   Paul Duke
 * @version  $Id: Table.java 29 2010-06-17 20:58:48Z mepcotterell $
 */

import static java.lang.System.out;
import java.io.Serializable;
import java.util.*;

/*****************************************************************************************
 * This class implements relational database tables (including attribute names, domains
 * and a list of tuples.  Five basic relational algebra operators are provided: project,
 * select, union, minus and join.  The insert data manipulation operator is also provided.
 * Missing are update and delete data manipulation operators.
 */
public class Table implements Serializable, Cloneable
{
    /** Counter for naming temporary tables.
     */
    private static int count = 0;

    /** Table name.
     */
    private final String name;

    /** Array of attribute names.
     */
    private final String [] attribute;

    /** Array of attribute domains: a domain may be
     *  boolean types: Boolean
     *  integer types: Long, Integer, Short, Byte
     *  real types: Double, Float
     *  string types: Character, String
     */
    private final Class [] domain;

    /** Collection of tuples (data storage).
     */
    private final List <Comparable []> tuples;

    /** Number of attributes forming the primary key (first keySize attributes)
     */
    private final int keySize;

    /** Index into tuples.
     */
    private final Map <KeyType, Comparable []> index;

    /*************************************************************************************
     * Construct an empty table from the meta-data specifications.
     * @param _name       the name of the relation
     * @param _attribute  the string containing attributes names
     * @param _domain     the string containing attribute domains (data types)
     * @param _keySize    the number of attributes forming the key
     */  
    public Table (String _name, String [] _attribute, Class [] _domain, int _keySize)
    {
        name      = _name;
        attribute = _attribute;
        domain    = _domain;
        keySize   = _keySize;
        tuples    = new ArrayList <Comparable []> ();         // also try LinkedList, FileList
        index     = new TreeMap <KeyType, Comparable []> ();  // also try HashMap, BPTreeMap, etc.
    } // Table

    /*************************************************************************************
     * Construct an empty table from the raw string specifications.
     * @param name        the name of the relation
     * @param attributes  the string containing attributes names
     * @param domains     the string containing attribute domains (data types)
     */
    public Table (String name, String attributes, String domains, int _keySize)
    {
        this (name, attributes.split (" "), findClass (domains.split (" ")), _keySize);

        out.println ("DDL> create table " + name + " (" + attributes + ")");
    } // Table

    /*************************************************************************************
     * Construct an empty table using the meta-data of an existing table.
     * @param tab     the table supplying the meta-data
     * @param suffix  the suffix appended to create new table name
     */
    public Table (Table tab, String suffix)
    {
        this (tab.name + suffix, tab.attribute, tab.domain, tab.keySize);
    } // Table

    /*************************************************************************************
     * Project the tuples onto a lower dimension by keeping only the given attributes.
     * #usage movie.project ("title year studioNo")
     * @param attributeList  the attributes to project onto
     * @return  a table of projected tuples
     * @author John Miller
     * @author Michael Cotterell
     */
    public Table project (String attributeList)
    {
        out.println ("RA> " + name + ".project (" + attributeList + ")");

        String [] pAttribute = attributeList.split (" ");
        Class []  colDomain  = extractDom (match (pAttribute), domain);
        int       tupSize    = pAttribute.length;
        int       kSize      = (keySize < tupSize) ? keySize : tupSize; 
        Table     result     = new Table (name + count++, pAttribute, colDomain, kSize);

        /**
         * @author Michael Cotterell
         */
        for (int i = 0; i < tuples.size (); i++) {

        	/** 
        	 * The tuple that we wish to project on to.
        	 */
        	Comparable [] currentTuple = tuples.get(i);
        	
        	/**
        	 * The tuple that results from the projection.
        	 */
        	Comparable [] projectedTuple = new Comparable[tupSize];
        	
        	/**
        	 * Match attribute names and project
        	 */
        	for (int j = 0; j < tupSize; j++) {
        		int k = 0;
        		while (pAttribute[j].compareTo(attribute[k]) != 0) k++;
        		projectedTuple[j] = currentTuple[k];
        	}
        	
        	/** 
        	 * Add the projected tuple to the result
        	 */
        	result.insert(projectedTuple);
        	
        } // for

        return result;
    } // project

    /*************************************************************************************
     * Select the tuples satisfying the given condition.
     * A condition is written as infix expression consists of 
     *   6 comparison operators: "==", "!=", "<", "<=", ">", ">="
     *   2 Boolean operators:    "&", "|"  (from high to low precedence)
     * #usage movie.select ("1979 < year & year < 1990")
     * @param condition  the check condition for tuples
     * @return  a table with tuples satisfying the condition
     */
    public Table select (String condition)
    {
        out.println ("RA> " + name + ".select (" + condition + ")");

        Table result = new Table (name + count++, attribute, domain, keySize);

        for (Comparable [] tup : tuples) {
            if (evalTup (condition, tup)) {
                result.tuples.add (tup);
            } // if
        } // for

        return result;
    } // select

    /**
     * Checks union compatability. 
     * #usage t1.unionCompatible(t2)
     * @param table2 the rhs table in the union compatibility test
     * @return a Boolean representing whether table2 is union compatible with the current table.
     * @author Michael Cotterell
     */
    public Boolean unionCompatible(Table table2) {
    	
    	/**
    	 * Tuple size check using domains
    	 */
    	if (this.domain.length != table2.domain.length) {
    		out.println("Tuple/Domain Size Mismatch");
    		return false;
    	}
    	
        /**
         * Tuple domains match check
         */
        for (int i = 0; i < this.domain.length; i++) {
        	if (this.domain[i] != table2.domain[i]) {
        		out.println("Domain Mismatch!");
            	return false;
        	}
        }
        
        /**
         * Tuple attribute names match check
         */
        for (int i = 0; i < this.domain.length; i++) {
        	if (this.attribute[i].compareTo(table2.attribute[i]) != 0) {
        		out.println("Attribute Name Mismatch!");
            	return false;
        	}
        }
        
        return true;
        
    }
    
    /*************************************************************************************
     * Union this table and table 2.
     * #usage movie.union (show)
     * @param table2  the rhs table in the union operation
     * @return  a table representing the union
     * @author Michael Cotterell
     */
    public Table union (Table table2)
    {
        out.println ("RA> " + name + ".union (" + table2.name + ")");

        Table result = new Table (name + count++, attribute, domain, keySize);
 
        /**
         * Check for union compatability
         */
        if (!this.unionCompatible(table2)) {
        	out.println("Not union compatible.");
        	return result;
        }
        
        /**
         * Copy all tuples from this table into the result.
         */
        for (Comparable [] tup : tuples) {
        	result.insert(tup);
        }
        
        /**
         * Copy all the tuples from table2 that aren't in this table into
         * the result.
         */
        for (Comparable [] tup : table2.tuples) {
        	boolean check = true;
        	for (Comparable [] cmp : tuples) {
        		if (tup == cmp) check = false;
        	}
        	if (check) result.insert(tup);
        }

        return result;
    } // union

    /*************************************************************************************
     * Take the difference of this table and table 2.
     * #usage movie.minus (show)
     * @param table2  The rhs table in the minus operation
     * @return  a table representing the difference
     * @author Paul Duke
     */
    public Table minus (Table table2)
    {
        out.println ("RA> " + name + ".minus (" + table2.name + ")");

        Table result = new Table (name + count++, attribute, domain, keySize);
        
        /**
         * Check for union compatability
         */
        if (!this.unionCompatible(table2)) {
        	out.println("Not union compatible.");
        	return result;
        }
        
        boolean inTable2;
        int currRowT1 = 0 , currRowT2 = 0;
        //for(currRowT1 = 0; currRowT1 < tuples.size(); currRowT1++)
        for(Comparable [] tuple : tuples)
        {
            inTable2 = false;
            for(Comparable [] compareTuple : table2.tuples)
            {
                if(tuple == compareTuple)
                {
                    inTable2 = true;
                    break;
                }// if

            }// for
            if(!inTable2)
            {
                result.insert(tuple);
                currRowT1++;
            }// if
        }// for
        return result;
    } // minus

    /*********************************************************************
     * Join this table and table 2.  If an attribute name appears in both
     * tables, assume it is from the first table unless it is qualified
     * with the first letter of the second table's name (e.g., "s.").
     * In the result, disambiguate the attribute names in a similar way
     * (e.g., prefix the second occurrence with "s_").
     * #usage movie.join ("studioNo == name", studio);
     * #usage movieStar.join ("name == s.name", starsIn);
     * @param condition  the join condition for tuples
     * @param table2     the rhs table in the join operation
     * @return  a table representing the join
     */
    public Table join (String condition, Table table2)
    {
        out.println ("RA> " + name + ".join (" + condition + ", " + table2.name + ")");

        Table result = new Table (name + count++, new String [0], new Class [0], 0);   // FIX

             //-----------------\\ 
            // TO BE IMPLEMENTED \\
           //---------------------\\ 

        return result;
    } // join

    /*************************************************************************************
     * Insert a tuple to the table.
     * #usage movie.insert ("'Star_Wars'", 1977, 124, "T", "Fox", 12345)
     * @param tuple    the array of attribute values
     * @return  whether insertion was successful
     */
    public boolean insert (Comparable [] tup)
    {
        out.println ("DML> insert into " + name + " values ( " + Arrays.toString (tup) + " )");

        if (typeCheck (tup)) {
            tuples.add (tup);
            index.put (new KeyType (tup, keySize), tup);
            return true;
        } else {
            return false;
        } // if
    } // insert

    /*************************************************************************************
     * Print the table.
     */
    public void print ()
    {
        out.println ("\n Table " + name);
        out.print ("|-");
        for (int i = 0; i < attribute.length; i++) out.print ("---------------");
        out.println ("-|");
        out.print ("| ");
        for (String a : attribute) out.printf ("%15s", a);
        out.println (" |");
        out.print ("|-");
        for (int i = 0; i < attribute.length; i++) out.print ("---------------");
        out.println ("-|");
        for (Comparable [] tup : tuples) {
            out.print ("| ");
            for (Comparable attr : tup) out.printf ("%15s", attr);
            out.println (" |");
        } // for
        out.print ("|-");
        for (int i = 0; i < attribute.length; i++) out.print ("---------------");
        out.println ("-|");
    } // print

    /*************************************************************************************
     * Get the column position for the given attribute name.
     * @param attr  the given attribute name
     * @return  a column position
     */
    private int columnPos (String attr)
    {
        for (int i = 0; i < attribute.length; i++) {
           if (attr.equals(attribute[i])) {
               return i;
           } // if
        } // for

        return -1;  // not found
    } // columnPos

    /*************************************************************************************
     * Check whether the tuple satisfies the condition.
     * @param condition  the infix expression for the condition
     * @param tup        the tuple to check
     * @return  whether to keep the tuple
     */
    private boolean evalTup (String condition, Comparable [] tup)
    {
        if (condition == null || condition.equals("")) {
            return true;
        } // if

        String [] postfix = infix2postfix (condition.split (" "));
        Stack<Object> s   = new Stack<Object> ();
           
        for (int i = 0; i < postfix.length; i++) {

        	boolean isOperator = true;
        	  	
        	if (postfix[i].equals("==")) {
        		
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = false;
        		
        		try {
        			double d1 = Double.parseDouble(op1.toString());
        			double d2 = Double.parseDouble(op2.toString());
        			result = (d1 == d2);
        		}// try
                        catch (NumberFormatException e) {
    				boolean check1 = false;
    				boolean check2 = false;
        			if ((op1.toString().toLowerCase().compareTo("true") == 0) | (op1.toString().toLowerCase().compareTo("false") == 0)) {
    					check1 = true;
    				}//if
        			if ((op2.toString().toLowerCase().compareTo("true") == 0) | (op2.toString().toLowerCase().compareTo("false") == 0)) {
    					check2 = true;
    				}// if
        			if (check1 && check2) {
        				boolean b1 = Boolean.parseBoolean(op1.toString());
        				boolean b2 = Boolean.parseBoolean(op2.toString());
        				result = (b1 == b2);
        			}// if
                                else {
        				result = (op1.toString().compareTo(op2.toString()) == 0);
        			}// else
        		}// catch
       
        		s.push(result);
        		
        	}// if
                else if (postfix[i].equals("!=")) {
        		
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = false;
        		
        		try {
        			double d1 = Double.parseDouble(op1.toString());
        			double d2 = Double.parseDouble(op2.toString());
        			result = (d1 != d2);
        		}// try
                        catch (NumberFormatException e) {
    				boolean check1 = false;
    				boolean check2 = false;
        			if ((op1.toString().toLowerCase().compareTo("true") == 0) | (op1.toString().toLowerCase().compareTo("false") == 0)) {
    					check1 = true;
    				}// if
        			if ((op2.toString().toLowerCase().compareTo("true") == 0) | (op2.toString().toLowerCase().compareTo("false") == 0)) {
    					check2 = true;
    				}// if
        			if (check1 && check2) {
        				boolean b1 = Boolean.parseBoolean(op1.toString());
        				boolean b2 = Boolean.parseBoolean(op2.toString());
        				result = (b1 != b2);
        			}// if
                                else {
        				result = (op1.toString().compareTo(op2.toString()) != 0);
        			}// else
        		}// catch
       
        		s.push(result);
        		 
        	}// if
                else if (postfix[i].equals("<")) {
        		
        		/**
        		 * Due to the order in which we pop things off the stack, we need to flip
        		 * the operator.
        		 */
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = false;
        		
        		try {
        			double d1 = Double.parseDouble(op1.toString());
        			double d2 = Double.parseDouble(op2.toString());
        			result = (d1 > d2);
        		}// try
                        catch (NumberFormatException e) {
        			result = (op1.toString().compareTo(op2.toString()) > 0); 			
        		}// catch
       
        		s.push(result);
        	
        	}// if
                else if (postfix[i].equals("<=")) {
        		
        		/**
        		 * Due to the order in which we pop things off the stack, we need to flip
        		 * the operator.
        		 */
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = false;
        		
        		try {
        			double d1 = Double.parseDouble(op1.toString());
        			double d2 = Double.parseDouble(op2.toString());
        			result = (d1 >= d2);
        		}// try
                        catch (NumberFormatException e) {
        			result = (op1.toString().compareTo(op2.toString()) >= 0); 			
        		}// catch
       
        		s.push(result);
        		
        	}// if
                else if (postfix[i].equals(">")) {
        		
        		/**
        		 * Due to the order in which we pop things off the stack, we need to flip
        		 * the operator.
        		 */
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = false;
        		
        		try {
        			double d1 = Double.parseDouble(op1.toString());
        			double d2 = Double.parseDouble(op2.toString());
        			result = (d1 < d2);
        		}// try
                        catch (NumberFormatException e) {
        			result = (op1.toString().compareTo(op2.toString()) < 0); 			
        		}// catch
       
        		s.push(result);
        	
        	}// if
                else if (postfix[i].equals(">=")) {
        		
        		/**
        		 * Due to the order in which we pop things off the stack, we need to flip
        		 * the operator.
        		 */
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = false;
        		
        		try {
        			double d1 = Double.parseDouble(op1.toString());
        			double d2 = Double.parseDouble(op2.toString());
        			result = (d1 <= d2);
        		}// try
                        catch (NumberFormatException e) {
        			result = (op1.toString().compareTo(op2.toString()) <= 0); 			
        		}// catch
       
        		s.push(result);
        		
        	}// if
                else if (postfix[i].equals("&")) {
        		
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = ((Boolean)op1 & (Boolean)op2);
        		s.push(result);
        	
        	}// if
                else if (postfix[i].equals("|")) {
        		
        		Object op1 = s.pop();
        		Object op2 = s.pop();
        		
        		boolean result = ((Boolean)op1 | (Boolean)op2);
        		s.push(result);
	
        	}// if
                else {
        		
        		isOperator = false;
       
        	}// else
        		
        	if (!isOperator) {
	        	
	        	boolean isAttributeName = false;
	        	
	        	for (int j = 0; j < this.attribute.length; j++) {
	        		if (postfix[i].compareTo(this.attribute[j]) == 0) {
	        			isAttributeName = true;
	        			s.push(tup[j]);
	        		}// if
	        	}// for
	        	
	        	if (!isAttributeName) {
	        		s.push(postfix[i]);
	        	}// if
	        	
        	}// if
        	
        }// for

        return (Boolean) s.pop ();
        
    } // evalTup

    /*************************************************************************************
     * Convert an infix expression to a postfix expression.  This implementation does not
     * handle parentheses.
     * @param infix  a tokenized infix expression
     * @return  a resultant tokenized postfix expression
     * @author Michael Cotterell
     * @author Paul Duke
     */
    private String [] infix2postfix (String [] infix)
    {
    	/**
    	 * List to store the generated postfix expression
    	 */
        List<String> postfix = new ArrayList<String>();
        
        /**
         * A stack used for operator precedence
         */
        Stack<String> operatorStack = new Stack<String>();
        
        /**
         * @todo Make the precedence map static so that it's not
         * created everytime infix2postfix is called.
         */
        
        /**
         * A map from operator -> precedence
         */
        Map<String, Integer> precedenceMap = new HashMap<String, Integer>();
        
        precedenceMap.put("&", 	1);
        precedenceMap.put("|", 	1);
        precedenceMap.put(">=", 2);
        precedenceMap.put(">", 	2);
        precedenceMap.put("<=", 2);
        precedenceMap.put("<", 	2);
        precedenceMap.put("!=", 2);
        precedenceMap.put("==", 2);
        
        // check the length of the expression
        if (infix.length != 0) {
        	
        	for (int i = 0; i < infix.length; i++) {
        		
        		// precedence is null for operands
        		Integer precedence = precedenceMap.get(infix[i]);

        		if (precedence != null) {
        			// then the current token is an operator
        			while (!operatorStack.isEmpty()) {
        				String opFromStack = operatorStack.pop();
          				if (precedenceMap.get(opFromStack) < precedence) {
        					operatorStack.push(opFromStack);
        					break;
        				} else {
        					postfix.add(opFromStack);
        				}      				
        			}
        			
        			operatorStack.push(infix[i]);
        			
        		} else { 
        			// current token is not an operator
        			postfix.add(infix[i]);
        		}
        		
        	}
        	
        	// add the remaining operators to the postfix expression
        	while (!operatorStack.isEmpty()) {
        		postfix.add(operatorStack.pop());
        	}
        	
        }
        
        return postfix.toArray(new String[postfix.size()]);
    } // infix2postfix

    /*************************************************************************************
     * Match the column and attribute names to determine the domains.
     * @param column  the array of column names
     * @return  an array of column index position
     */
    private int [] match (String [] column)
    {
        int [] colPos = new int [column.length];

        for (int j = 0; j < column.length; j++) {
            boolean matched = false;
            for (int k = 0; k < attribute.length; k++) {
                if (column [j].equals (attribute [k])) {
                    matched = true;
                    colPos [j] = k;
                } // for
            } // for
            if ( ! matched) {
                out.println ("match: domain not found for " + column [j]);
            } // if
        } // for

        return colPos;
    } // match

    /*************************************************************************************
     * Check the size of the tuple (number of elements in list) as well as the type of
     * each value to ensure it is from the right domain. 
     * @param tup  The tuple as a list of attribute values
     * @return  whether the tuple has the right size and values that comply
     *          with the given domains
     * @author Michael Cotterell
     * @author Paul Duke
     */
    private boolean typeCheck (Comparable [] tup)
    { 
    	boolean check = true;
    	if (tup.length != this.domain.length) {
    		check = false;
    	} else {
	    	for (int i = 0; i < this.domain.length; i++) {
	    		if (!this.domain[i].isInstance(tup[i])) {
	    			check = false;
	    			break;
	            }
	        }
    	}
        return check;
    } // typeCheck
    
    

    /*************************************************************************************
     * Find the classes in the "java.lang" package with given names.
     * @param className  the array of class name (e.g., {"Integer", "String"})
     * @return  an array of Java classes
     */
    private static Class [] findClass (String [] className)
    {
        Class [] classArray = new Class [className.length];

        for (int i = 0; i < className.length; i++) {
            try {
                classArray [i] = Class.forName ("java.lang." + className [i]);
            } catch (ClassNotFoundException ex) {
                out.println ("findClass: " + ex);
            } // try
        } // for

        return classArray;
    } // findClass

    /*************************************************************************************
     * Extract the corresponding domains.
     * @param colPos the column positions to extract.
     * @param group  WHere to extract from
     * @return  the extracted domains
     */
    private Class [] extractDom (int [] colPos, Class [] group)
    {
        Class [] obj = new Class [colPos.length];

        for (int j = 0; j < colPos.length; j++) {
            obj [j] = group [colPos [j]];
        } // for

        return obj;
    } // extractDom
    
    /*************************************************************************************
     * The main method is used for testing purposes only.
     * @param args  the command-line arguments
     * @author Michael Cotterell
     */
    public static void main (String [] args)
    {
 
    	// create some tables
    	Table customer = new Table("customer", "id name age email", "Integer String Integer String", 1);
    	Table customer2 = new Table("customer2", "id name age email", "Integer String Integer String", 1);
    	
    	// setup dummy data
    	List<Comparable []> tups = new ArrayList<Comparable []>();
    	tups.add(new Comparable[] {1, "Michael", 22, "mc"});
    	tups.add(new Comparable[] {2, "Paul", 21, "pd"});
    	
    	List<Comparable []> tups2 = new ArrayList<Comparable []>();
    	tups2.add(new Comparable[] {3, "John", 32, "jd"});
    	tups2.add(new Comparable[] {4, "Smith", 18, "sh"});
    	
    	// insert tups
    	for (Comparable [] tup : tups) {
    		customer.insert(tup);
    	}
    	
    	for (Comparable [] tup : tups2) {
    		customer2.insert(tup);
    	}
    	
    	// print the table
    	customer.print();
    	customer2.print();
    	
    	// test project
    	out.println("\n|--------------------------------------------------------------|");
    	out.println(" TESTING Table.project()");
    	out.println("|--------------------------------------------------------------|");
    	customer.project("id name").print();
    	customer.project("id").print();
    	customer2.project("name email").print();
    	customer2.project("email").print();
    	
    	out.println("\n|--------------------------------------------------------------|");
    	out.println(" TESTING Table.select()");
    	out.println("|--------------------------------------------------------------|");
    	customer.select("age == 21.0").print();
    	customer.select("name == Michael").print();
    	customer.select("name == Paul").print();
    	customer.select("name != Paul").print();
    	customer.select("name != Michael").print();
    	customer.select("age == 22").print();
    	customer.select("age < 20").print();
    	customer.select("age <= 22").print();
    	customer.select("age > 20").print();
    	customer.select("age <= 22.0").print();
    	customer.select("age >= 21.0").print(); // (21 >= 21.0) -> false
    	
    	out.println("\n|--------------------------------------------------------------|");
    	out.println(" TESTING Table.union()");
    	out.println("|--------------------------------------------------------------|");
    	customer.union(customer2).print();
    	
    	out.println("\n|--------------------------------------------------------------|");
    	out.println(" TESTING Table.minus()");
    	out.println("|--------------------------------------------------------------|");
    	
    	customer.union(customer2).minus(customer2).print(); // should result in just customer
    	customer.union(customer2).minus(customer).print(); // should result in just customer2
    	
    	out.println("\n|--------------------------------------------------------------|");
    	out.println(" TESTING Table.select() with booleans");
    	out.println("|--------------------------------------------------------------|");
    	customer.union(customer2).select("name == Michael | age == 21").print();
    	customer.union(customer2).select("true == true").print();
    	
    }

} // Table class
